package com.swak.p2p.cluster;

import java.util.concurrent.CompletionStage;

import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.Cluster;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandlerAdapter;
import com.swak.reactivex.transport.Message;

public class ClusterA extends ExchangeHandlerAdapter {

	private Member member;

	public ClusterA() throws RemotingException {
		member = Cluster.join("Raid", this);
	}

	// ***************** 回调事件处理 **************************
	@Override
	public void connected(Channel channel) throws RemotingException {
		System.out.println("-- 建立连接 --");
		System.out.println(channel.url());
		System.out.println(channel.localAddress() + " " + channel.remoteAddress());
		System.out.println("------------");
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
		System.out.println("-- 断开连接 --：" + channel.url());
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		System.out.println("收到消息：" + ((Message) message).getData());
	}

	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		return super.reply(channel, message);
	}

	// ***************** 主动的发送数据 **************************

	public void send(Object message) throws RemotingException {
		//this.member.send(message);
		this.member.broadcast(message);
	}

	public static void main(String[] args) throws RemotingException, InterruptedException {
		ClusterA cluster = new ClusterA();
		Thread.sleep(10000);
		cluster.send("Hello!");
	}
}
